package com.openness.spider.zk.lock;

import java.util.Comparator;
import java.util.List;
import java.util.NavigableSet;
import java.util.TreeSet;

import org.apache.zookeeper.ZooKeeper;

import com.openness.spider.commons.Statics;
import com.openness.spider.zk.ProtocolSupport;
import com.openness.spider.zk.ZooKeeperOperation;

/**
 * @author Rain
 * @version 2013-6-19 下午9:31:36
 */

public class DistributedLock extends ProtocolSupport {

	private ZooKeeper zk;

	private String node;

	private String lock;

	private static final String SEPERATOR = "/";

	private static final String PREFIX = "lock_";

	public DistributedLock(ZooKeeper zk, String node) {
		this.zk = zk;

		this.node = node;
	}

	private NavigableSet<String> orderChild(List<String> child) {
		NavigableSet<String> orderedChild = new TreeSet<String>(
				new Comparator<String>() {

					@Override
					public int compare(String str1, String str2) {
						int begin1 = str1.lastIndexOf('_') + 1;
						int begin2 = str2.lastIndexOf('_') + 1;

						long seq1 = Long.valueOf(str1.substring(begin1));
						long seq2 = Long.valueOf(str2.substring(begin2));

						return seq1 < seq2 ? -1 : 1;
					}

				});

		for (String children : child) {
			orderedChild.add(children);
		}

		return orderedChild;
	}

	public void lock() throws Exception {
		lock = retryOperation(new ZooKeeperOperation<String>() {

			@Override
			public String execute() throws Exception {
				return zk.create(node + SEPERATOR + PREFIX, Statics.EMPTY_BYTEARRAY, Statics.ACL,
						Statics.CREATEMODE_EPHEMERAL_SEQUENTIAL);
			}

		});

		while (true) {
			List<String> child = retryOperation(new ZooKeeperOperation<List<String>>() {

				@Override
				public List<String> execute() throws Exception {
					return zk.getChildren(node, false);
				}

			});

			NavigableSet<String> orderedChild = orderChild(child);

			String lowest = node + SEPERATOR + orderedChild.first();

			if (lowest.equals(lock)) {
				break;
			}

			final String nextLowest = node + SEPERATOR
					+ orderedChild.lower(lowest);

			final WatcherLatch latch = new WatcherLatch();

			boolean exist = retryOperation(new ZooKeeperOperation<Boolean>() {

				@Override
				public Boolean execute() throws Exception {
					return zk.exists(nextLowest, latch) != null ? true : false;
				}

			});

			if (exist) {
				latch.await();
			}
		}
	}

	public void unlock() throws Exception {
		retryOperation(new ZooKeeperOperation<Object>() {

			@Override
			public Object execute() throws Exception {
				zk.delete(lock, -1);

				return null;
			}

		});
	}

}
